#include "access/ringbuffer.h"
#include <string.h>
#include "access/xlogrecord.h"
/**
 * @file
 * Implementation of ring buffer functions.
 */

void ring_buffer_init(ring_buffer_t *buffer, wal_batch_t *buf, size_t buf_size) {
  RING_BUFFER_ASSERT(RING_BUFFER_IS_POWER_OF_TWO(buf_size) == 1);
  SpinLockInit(&buffer->mutex);
  buffer->buffer = buf;
  buffer->buffer_mask = buf_size - 1;
  buffer->tail_index = 0;
  buffer->head_index = 0;
}

wal_batch_t *ring_buffer_queue(ring_buffer_t *buffer, wal_batch_t data) {
  wal_batch_t* curWal = NULL;
  SpinLockAcquire(&buffer->mutex);
  /* Is buffer full? */
  if(ring_buffer_is_full(buffer)) {
  	SpinLockRelease(&buffer->mutex);
    return NULL;
  }
  buffer->buffer[buffer->head_index].startLsn  = data.startLsn;
  buffer->buffer[buffer->head_index].endLsn = data.endLsn;
  buffer->buffer[buffer->head_index].checkPointLsn = data.checkPointLsn;
  pg_atomic_exchange_u32(&buffer->buffer[buffer->head_index].status,(uint32_t)UNKOWNSTATUS);
  curWal = &buffer->buffer[buffer->head_index];
  buffer->head_index = ((buffer->head_index + 1) & RING_BUFFER_MASK(buffer));
  SpinLockRelease(&buffer->mutex);
  return curWal;
}

uint8_t ring_buffer_dequeue(ring_buffer_t *buffer, wal_batch_t *data) {
  SpinLockAcquire(&buffer->mutex);
  if(ring_buffer_is_empty(buffer)) {
    /* No items */
  	SpinLockRelease(&buffer->mutex);
    return 0;
  }
  if (data != NULL) {
  	*data = buffer->buffer[buffer->tail_index];
  }
  buffer->tail_index = ((buffer->tail_index + 1) & RING_BUFFER_MASK(buffer));
  SpinLockRelease(&buffer->mutex);
  return 1;
}

uint8_t ring_buffer_dequeue_arr(ring_buffer_t *buffer, uint32 size) {
  SpinLockAcquire(&buffer->mutex);
  if(ring_buffer_is_empty(buffer)) {
    /* No items */
  	SpinLockRelease(&buffer->mutex);
    return 0;
  }
  ring_buffer_size_t pos = buffer->tail_index;
  for(uint32 i = 0;i<size;i++) {
  	pg_atomic_exchange_u32(&buffer->buffer[pos].status,(uint32_t)UNKOWNSTATUS);
  	pos = ((pos+1) & RING_BUFFER_MASK(buffer));
  }
  buffer->tail_index = ((buffer->tail_index + size) & RING_BUFFER_MASK(buffer));
  SpinLockRelease(&buffer->mutex);
  return 1;
}


uint8_t ring_buffer_peek(ring_buffer_t *buffer, wal_batch_t **data, ring_buffer_size_t index) {
  SpinLockAcquire(&buffer->mutex);
  if(index >= ring_buffer_num_items(buffer)) {
    /* No items at index */
    SpinLockRelease(&buffer->mutex);
    return 0;
  }
  /* Add index to pointer */
  ring_buffer_size_t data_index = ((buffer->tail_index + index) & RING_BUFFER_MASK(buffer));
  *data = &buffer->buffer[data_index];
  SpinLockRelease(&buffer->mutex);
  return 1;
}

uint8_t ring_buffer_will_full(ring_buffer_t *buffer) {
  ring_buffer_size_t num = ((buffer->head_index - buffer->tail_index) & RING_BUFFER_MASK(buffer));
  return num > 0.9 * RING_BUFFER_MASK(buffer);
}

ring_buffer_t* gRingBufferManger;
Size WalReadBufferShmemSize(void) {
	Size		size;
	size = 0;
	size = add_size(size,sizeof(ring_buffer_t));
	//spaceNum of numbers wal batchs to manage wal buffer 
	size = add_size(size, spaceNum * sizeof(wal_batch_t));
	//256MB cache for wal parallel read
	size = add_size(size, spaceNum * 4 * XLOG_BLCKSZ);
	return size;
}

// one elem max receive no more than 32k,64MB = spaceNum * 4 * XLOG_BLCKSZ
const int spaceNum = 8192;
void InitRingBufferSpace(void) {
	//default 256MB for cache
	char* gFreeSpace = NULL;
	bool		found;
	gFreeSpace = (char *)
		ShmemInitStruct("walreadbuffer",
						sizeof(ring_buffer_t) + spaceNum * sizeof(wal_batch_t) + spaceNum * 4 * XLOG_BLCKSZ,
						&found);
	
	if (gFreeSpace == NULL) {
		elog(FATAL,"gFreeSpace malloc failed");
	}
	gRingBufferManger = (ring_buffer_t*)gFreeSpace;
	gFreeSpace += sizeof(ring_buffer_t);
	wal_batch_t* gManageFreeList;
	gManageFreeList = (wal_batch_t*)gFreeSpace ;
	gFreeSpace += spaceNum * sizeof(wal_batch_t);
	int i = 0;
	for(;i<spaceNum; i++) {
		gManageFreeList[i].data = &gFreeSpace[i*4*XLOG_BLCKSZ];
	}
	ring_buffer_init(gRingBufferManger,gManageFreeList,spaceNum);
}

int walRecordQuery(char**buffer,int* curpos,int* maxspace,uint64 lsn) {
	ring_buffer_size_t maxIdx = gRingBufferManger->maxIdx;
	if (maxIdx == 0) {
		return -1;
	}
	ring_buffer_size_t tailIdx = gRingBufferManger->tail_index; 
	int low = tailIdx,high = ((tailIdx+maxIdx) & RING_BUFFER_MASK(gRingBufferManger)), mid = 0;
	if (low > high) {
		if (gRingBufferManger->buffer[gRingBufferManger->buffer_mask].startLsn + gRingBufferManger->buffer[gRingBufferManger->buffer_mask].dataLen > lsn) {
			high = gRingBufferManger->buffer_mask+1;
		} else {
			low = 0;
		}
	}
	if (gRingBufferManger->buffer[high-1].startLsn == 0) {
		high -= 2;
	} else {
		high -=1;
	}
	bool find = false;
	while(low <= high) {
		mid = (low + high) / 2;
		if (gRingBufferManger->buffer[mid].startLsn > lsn) {
			high = mid - 1;
		} else if (gRingBufferManger->buffer[mid].startLsn < lsn) {		
			low = mid + 1;
		} else {
			find = true;
			break;
		}
	}
	XLogRecord* record = NULL;
	int xllen = -1;
	bool  extandFlag = false;
	if (find == true) {
		record = (XLogRecord*)gRingBufferManger->buffer[mid].data;
		xllen = record->xl_tot_len;
	} else {
		record = (XLogRecord*)gRingBufferManger->buffer[high].data;
		if (gRingBufferManger->buffer[high].startLsn + gRingBufferManger->buffer[high].dataLen <= lsn) {
			return -1;
		} else {
			record = (XLogRecord*)(gRingBufferManger->buffer[high].data + (lsn-gRingBufferManger->buffer[high].startLsn));
			xllen = record->xl_tot_len;
		}
	}
	if (xllen != -1) {
		while (*curpos + xllen > *maxspace) {
			*maxspace += *maxspace;
			extandFlag = true;
		}
		if (extandFlag == true) {
			char* ptr = malloc(*maxspace);
			memcpy(ptr,*buffer,*curpos);
			free(*buffer);
			*buffer = ptr;
		}
		memcpy(*buffer+*curpos,record,xllen);
		*curpos += xllen;
	}
	return xllen;
}


extern inline uint8_t ring_buffer_is_empty(ring_buffer_t *buffer);
extern inline uint8_t ring_buffer_is_full(ring_buffer_t *buffer);
extern inline ring_buffer_size_t ring_buffer_num_items(ring_buffer_t *buffer);

